/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.dw.lidejia.hadooptotalsort.meta;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableUtils;

/**
 *
 * @author lidejia
 */
public class SortKeyMeta {

    private static interface MetaHandler {

        public int getLength(byte[] buffer, int start);

        public WritableComparable parse(String token);
    }

    private static class IntMetaHandler implements MetaHandler {

        @Override
        public int getLength(byte[] buffer, int start) {
            return 4;
        }

        @Override
        public WritableComparable parse(String token) {
            return new IntWritable(Integer.parseInt(token));
        }
    }

    private static class LongMetaHandler implements MetaHandler {

        @Override
        public int getLength(byte[] buffer, int start) {
            return 8;
        }

        @Override
        public WritableComparable parse(String token) {
            return new LongWritable(Long.parseLong(token));
        }
    }

    private static class TextMetaHandler implements MetaHandler {

        @Override
        public int getLength(byte[] buffer, int start) {
            int length = 0;
            try {
                length = WritableUtils.decodeVIntSize(buffer[start]) + WritableComparator.readVInt(buffer, start);
            } catch (IOException ex) {
                Logger.getLogger(SortKeyMeta.class.getName()).log(Level.SEVERE, null, ex);
            }
            return length;
        }

        @Override
        public WritableComparable parse(String token) {
            return new Text(token);
        }
    }
    //
    //
    private static final Map<Class<? extends WritableComparable>, MetaHandler> handlers = new HashMap<Class<? extends WritableComparable>, MetaHandler>();

    static {
        handlers.put(IntWritable.class, new IntMetaHandler());
        handlers.put(LongWritable.class, new LongMetaHandler());
        handlers.put(Text.class, new TextMetaHandler());
    }
    //
    private Class<? extends WritableComparable> type;
    private int index;
    private boolean orderReverse;
    //
    private MetaHandler handler;

    public SortKeyMeta(Class<? extends WritableComparable> type, int index, boolean orderReverse) {
        this.type = type;
        this.index = index;
        this.orderReverse = orderReverse;
        this.handler = handlers.get(type);
    }

    public WritableComparable createInstance() {
        WritableComparable instance = null;
        try {
            instance = type.newInstance();
        } catch (InstantiationException ex) {
            Logger.getLogger(SortKeyMeta.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IllegalAccessException ex) {
            Logger.getLogger(SortKeyMeta.class.getName()).log(Level.SEVERE, null, ex);
        }
        return instance;
    }

    public int compare(WritableComparable first, WritableComparable second) {
        int cmp = first.compareTo(second);
        if (orderReverse) {
            cmp = -cmp;
        }
        return cmp;
    }

    public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
        int cmp = WritableComparator.get(type).compare(b1, s1, l1, b2, s2, l2);
        if (orderReverse) {
            cmp = -cmp;
        }
        return cmp;
    }

    public int getLength(byte[] buffer, int start) {
        return handler.getLength(buffer, start);
    }

    public WritableComparable parse(String[] tokens) {
        String token = tokens[index];
        return handler.parse(token);
    }
}
